/**
 * @file DataBase.cpp
 * @Synopsis
 * @author Mingyu Wu
 * @version 1.0.0
 * @date 2024-04-20
 */
#include "DataBase.h"

bool                          DataBase::online = false;
std::unique_ptr<RedisCluster> DataBase::cluster;

int DataBase::recvStreamCount = 0;

DataBase::DataBase() {
    zlog_w("初始化database");
    std::vector<Configuration::ClusterConf>::iterator iter;

    for (iter = Configuration::clusterConf.begin();
         iter != Configuration::clusterConf.end(); iter++) {
        ConnectionOptions option;
        option.host           = iter->Host;
        option.port           = iter->Port;
        option.password       = iter->Password;
        option.db             = 0;
        option.socket_timeout = std::chrono::seconds(5);
        connection_options.push_back(option);
    }
    zlog_i("载入配置文件集群信息个数:%ld", connection_options.size());

    zlog_w("< 载入配置信息个数: %ld >", Configuration::clusterConf.size());

    zlog_w("< Hash Table Count: %d >",
           Configuration::dataConf.begin()->HashGroupCount);

    zlog_w("< Data Sync Period: %d >",
           Configuration::dataConf.begin()->DataSyncUpdateInterval);

#if 0
    // redis connection initialize
    connection_options[0].port =
        NODE_1_PORT; // Optional. The default port is 6379.
    connection_options[0].host = NODE_1_HOST; // Required.
    connection_options[0].password =
        NODE_1_PASSWORD;          // Optional. No password by default.
    connection_options[0].db = 0; // Optional. Use the 0th database by default.
    connection_options[0].socket_timeout = std::chrono::seconds(10);
#endif

    pool_options.size = 3;
    // Optional. Max time to wait for a connection. 0ms by default, which means
    // wait forever. Say, the pool size is 3, while 4 threds try to fetch the
    // connection, one of them will be blocked.
    pool_options.wait_timeout = std::chrono::milliseconds(100);

    // Optional. Max lifetime of a connection. 0ms by default, which means never
    // expire the connection. If the connection has been created for a long
    // time, i.e. more than `connection_lifetime`, it will be expired and
    // reconnected.
    pool_options.connection_lifetime = std::chrono::minutes(10);
    this->GetID();
}

DataBase::~DataBase() {
}

void DataBase::CreatePingCommand(std::string &command, std::string host,
                                 int port) {
    // std::string command = "redis-cli -h 127.0.0.1 -p 7001 -c ping";
    command.clear();
    command = "redis-cli -h ";
    command.append(host);
    command.append(" -p ");
    command.append(std::to_string(port));
    command.append(" -c ping");
}

void DataBase::CreateClusterCommand(std::string &command, std::string host,
                                    int port) {
    command.clear();
    command = "redis-cli -h ";
    command.append(host);
    command.append(" -p ");
    command.append(std::to_string(port));
    command.append(" -c cluster info");
}

int DataBase::SendCommand(std::string command) {
    // 执行命令并打开输出流
    FILE *pipe = popen(command.c_str(), "r");
    if (!pipe) {
        zlog_w("fail to open pipe");
        return ErrorInfo::REDIS_CONNECT;
    }

    // 读取命令输出
    char        buffer[128];
    std::string result = "";

    if (fgets(buffer, 128, pipe) != nullptr) {
        result += buffer;
    }
    fclose(pipe);

    if (result.substr(0, 4) == "PONG") {
        return ErrorInfo::OK;
    }
    if (result.substr(0, 16) == "cluster_state:ok") {
        return ErrorInfo::OK;
    }
    return ErrorInfo::REDIS_CONNECT;
}

int DataBase::ConnectCluster() {
    int ret = ErrorInfo::OK;
    int i;
    if (this->online) {
        return ErrorInfo::OK;
    }

    zlog_w("data try to connect server");
    // 检查节点可用
    for (i = 0; i < 6; i++) {
        online = false;
        zlog_w("try node:%d", i + 1);
        ret = this->ConnectNode(i);
        if (ErrorInfo::OK == ret) {
            return ret;
        }
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    return ErrorInfo::REDIS_CONNECT;
}

int DataBase::ConnectNode(int index) {
    std::string command;
    int         ret = ErrorInfo::OK;
    CreatePingCommand(command, connection_options[index].host,
                      connection_options[index].port);
    ret = this->SendCommand(command);
    if (ErrorInfo::OK == ret) {
        zlog_w("[%02d] try: %s %d cluster server connected", index + 1,
               connection_options[index].host.c_str(),
               connection_options[index].port);
        // 每个节点尝试3次机会
        for (int j = 0; j < 3; j++) {
            ret = this->CheckClusterStatus(index);
            if (ret == ErrorInfo::OK) {
                zlog_w("[%02d] try: %s %d cluster state ok ", j + 1,
                       connection_options[index].host.c_str(),
                       connection_options[index].port);
                return ret;
            } else {
                zlog_w("[%02d] try: %s %d cluster state fail ", j + 1,
                       connection_options[index].host.c_str(),
                       connection_options[index].port);
            }
            // 防止redis集群结构调整,误报失联
            std::this_thread::sleep_for(std::chrono::seconds(3));
        }
    } else {
        zlog_w("%s %d cluster server is down",
               connection_options[index].host.c_str(),
               connection_options[index].port);
    }
    return ErrorInfo::REDIS_CONNECT;
}

int DataBase::CreateConnection(ConnectionOptions option) {
    for (int i = 0; i < 5; i++) {
        try {
            online = false;
            zlog_w("[%02d] try connect %d", i + 1, option.port);
            cluster = std::make_unique<RedisCluster>(option, pool_options);
            this->CleanStream();
            online = true;
            return ErrorInfo::OK;
        } catch (const Error &err) {
            zlog_w("%s", err.what());
            return ErrorInfo::REDIS_CONNECT;
        }
        std::this_thread::sleep_for(std::chrono::seconds(2));
    }
    return ErrorInfo::REDIS_CONNECT;
}

int DataBase::CheckClusterStatus(int index) {
    std::string command;
    int         ret = ErrorInfo::OK;
    // check cluster status
    CreateClusterCommand(command, connection_options[index].host,
                         connection_options[index].port);
    ret = this->SendCommand(command);

    if (ErrorInfo::OK == ret) {
        ret = this->CreateConnection(connection_options[index]);
        return ret;
    }
    return ErrorInfo::REDIS_CONNECT;
}

int DataBase::Init() {
    int ret = ErrorInfo::OK;
    ret     = this->ConnectCluster();

    if (ErrorInfo::OK != ret) {
        return ret;
    }
    // this->CleanStream();
    return ErrorInfo::OK;
}

bool DataBase::IsServerOnline() {
    return online;
}

int DataBase::UploadHandler(
    const std::string                                   hashKey,
    const std::unordered_map<std::string, std::string> &data) {
    int ret = ErrorInfo::OK;
    if (!this->IsServerOnline()) {
        zlog_w("cluster is down try to reconnect");
        ret = this->ConnectCluster();
        if (ErrorInfo::OK != ret) {
            return ret;
        }
    }

    try {
        cluster->hmset(hashKey, data.begin(), data.end());
        cluster->expire(hashKey,
                        Configuration::dataConf.begin()->DataExpireTime);
    } catch (const Error &err) {
        zlog_w("%s", err.what());
        return ErrorInfo::UPLOAD_FAIL;
    }
    return ErrorInfo::OK;
}

int DataBase::StreamRecv(
    std::unordered_map<std::string, std::string> &controlData) {
    //                                         field    value
    using Attrs      = std::vector<std::pair<std::string, std::string>>;
    using Item       = std::pair<std::string, Optional<Attrs>>;
    using ItemStream = std::vector<Item>;

    std::unordered_map<std::string, ItemStream>::iterator iterStream;
    ItemStream::iterator                                  iterIterm;
    Attrs::iterator                                       iterAttr;

    controlData.clear();

    // stream data        "key"
    std::unordered_map<std::string, ItemStream> result;

    // fetch data from redis
    try {
        cluster->xread(Configuration::dataConf.begin()->StreamKey, streamID,
                       std::chrono::seconds(overtime), 100,
                       std::inserter(result, result.end()));
        this->GetID();

        if (0 == result.size()) {
            zlog_w("stream 等待超时");
            return ErrorInfo::REDIS_OVERTIME;
        }

        for (iterStream = result.begin(); iterStream != result.end();
             iterStream++) {
            recvStreamCount += iterStream->second.size();

            for (iterIterm = iterStream->second.begin();
                 iterIterm != iterStream->second.end(); iterIterm++) {
                // 收到消息后,将消息从redis移除
                cluster->xdel(Configuration::dataConf.begin()->StreamKey,
                              iterIterm->first);

                for (iterAttr = iterIterm->second->begin();
                     iterAttr != iterIterm->second->end(); iterAttr++) {
                    controlData.emplace(iterAttr->first, iterAttr->second);
                }
            }
        }

    } catch (const TimeoutError &err) {
        zlog_w("streamKey: %s, %s, overtime",
               Configuration::dataConf.begin()->StreamKey.c_str(), err.what());

    } catch (const Error &err) {
        zlog_w("streamKey: %s recv fail ,%s",
               Configuration::dataConf.begin()->StreamKey.c_str(), err.what());
        return ErrorInfo::REDIS_STREAM_RECV;
    }
    return ErrorInfo::OK;
}

void DataBase::GetID() {
    struct timeval te;
    gettimeofday(&te, NULL); // 获取当前时间
    long long milliseconds =
        te.tv_sec * 1000LL +
        te.tv_usec / 1000; // 将秒转换为毫秒，并加上微秒的部分
    streamID = std::to_string(milliseconds - STREAM_ID_OFFSET);
}

void DataBase::CleanStream() {
    try {
        cluster->xtrim(Configuration::dataConf.begin()->StreamKey, 0, false);
        zlog_w("Stream 初始化");
    } catch (const Error &err) {
        zlog_w("清理消息队列失败");
    }
}

int DataBase::QueryData(std::string                                   hashKey,
                        std::unordered_map<std::string, std::string> &data) {

    if (!this->IsServerOnline()) {
        return ErrorInfo::REDIS_OFFLINE;
    }

    try {
        data.clear();
        cluster->hgetall(hashKey, std::inserter(data, data.begin()));

        // zlog_i("key: %s, Hash Tab 获取数据 %ld个", hashKey.c_str(),
        //        data.size());

    } catch (const Error &err) {
        zlog_w("%s", err.what());
        return ErrorInfo::REDIS_QUERY_FAIL;
    }

    return ErrorInfo::OK;
}

int DataBase::StreamSend(
    std::string                                       streamKey,
    std::vector<std::pair<std::string, std::string>> &controlData) {

    try {
        cluster->xadd(streamKey, "*", controlData.begin(), controlData.end());

    } catch (const Error &err) {
        zlog_w("%s", err.what());
        return ErrorInfo::REDIS_STREAM_SEND;
    }

    return ErrorInfo::OK;
}
